package main

import (
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"io/ioutil"
	"time"
)

var messagePubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
	fmt.Printf("messagePubHandler Received message: %s from topic: %s\n", msg.Payload(), msg.Topic())
}

var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {
	fmt.Println("Connected-client", client)
}

var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
	fmt.Printf("Connect lost: %v\n", err.Error())
}

var (

	// topName = "topic/test"
	/*var broker = "q5b4a411.ala.cn-hangzhou.emqxsl.cn"
	var port = 8883*/

	evchan = make(chan []byte, 4)
	broker = "mq.heatwatch.cn"
	port   = 1983

	/*username   = "test1"
	password   = "123"
	clientId   = "111111"
	pubTopName = "a/b/c"
	subTopName = "a/b/#"*/

	/*username = "kptmqtt"
	password = "kepaiteng"
	clientId = "ceshi"
	topName  = "ceshi/kptmqtt/ceshi/ceshi"*/

	username   = "kptmqtt"
	password   = "kepaiteng"
	clientId   = "ryjt"
	subTopName = "tmr/ryjt/kptmqtt/ryjt/pycx"
	pubTopName = "tmr/ryjt/kptmqtt/ryjt/pycx"

	// 05:27:14.711 [error] can't execute query <<"SELECT publish_acl, subscribe_acl FROM vmq_auth_acl WHERE  mountpoint=? AND client_id=? AND  username=? AND  password=MD5(?)">> due to exit:{{failed_to_prepare_statement,"Table 'mqttdb.vmq_auth_acl' doesn't exist"},{}}
	// 05:27:14.711 [error] can't call function auth_on_register with args [{addr,<<"27.115.106.2">>},{port,56001},{mountpoint,<<>>},{client_id,<<"111111">>},{username,<<"test">>},{password,<<"123">>},{clean_session,true}] in "./share/lua/auth/mysql.lua" due to {badarg,execute_equery,[<<"auth_mysql">>,<<"SELECT publish_acl, subscribe_acl FROM vmq_auth_acl WHERE  mountpoint=? AND client_id=? AND username=? AND password=MD5(?)">>,<<>>,<<"111111">>,<<"test">>,<<"123">>]}
)

func NewMqtt() {
	opts := mqtt.NewClientOptions()
	opts.AddBroker(fmt.Sprintf("tcp://%s:%d", broker, port))
	opts.SetClientID(clientId)
	opts.SetUsername(username)
	opts.SetPassword(password)
	opts.SetDefaultPublishHandler(messagePubHandler)
	opts.OnConnect = connectHandler
	opts.OnConnectionLost = connectLostHandler
	client := mqtt.NewClient(opts)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		panic(token.Error())
	}

	c, _ := json.Marshal(client)
	fmt.Println("=====client=====", string(c))

	go sub(client)
	MsgPrint()
	//publish(client)
}

func NewTlsConfig() *tls.Config {
	certpool := x509.NewCertPool()
	ca, err := ioutil.ReadFile("ca.pem")
	if err != nil {
		panic(err)
	}
	certpool.AppendCertsFromPEM(ca)
	clientKeyPair, err := tls.LoadX509KeyPair("client-crt.pem", "client-key.pem")
	if err != nil {
		panic(err)
	}
	return &tls.Config{
		Certificates:       []tls.Certificate{clientKeyPair},
		RootCAs:            certpool,
		ClientAuth:         tls.NoClientCert,
		ClientCAs:          nil,
		InsecureSkipVerify: true,
	}
}

func NewNotTlsConfig() *tls.Config {
	certpool := x509.NewCertPool()
	ca, err := ioutil.ReadFile("ca.pem")
	if err != nil {
		panic(err)
	}
	certpool.AppendCertsFromPEM(ca)
	return &tls.Config{
		RootCAs: certpool,
	}

}

func sub(client mqtt.Client) {
	b, _ := json.Marshal(client)
	fmt.Println("=====sub=====", string(b))
	client.Subscribe(subTopName, 1, func(client mqtt.Client, msg mqtt.Message) {
		fmt.Println("=====msg.Payload=====", string(msg.Payload()))
		evchan <- msg.Payload()
	})
}

func publish(client mqtt.Client) {
	num := 1000
	for i := 0; i < num; i++ {
		text := fmt.Sprintf("Message %d", i)
		token := client.Publish(pubTopName, 1, false, text)
		token.Wait()

		fmt.Println("=====publish=====", text)
		time.Sleep(time.Second)
	}
}

func MsgPrint() {
	for {
		select {
		case msg := <-evchan:
			fmt.Println(fmt.Sprintf("MsgPrint: %s\n", string(msg)))
		}
	}
}
